
package com.greencat.gwt.comet.server.impl;

import com.google.gwt.rpc.server.ClientOracle;
import com.google.gwt.user.server.rpc.SerializationPolicy;
import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpSession;
import com.greencat.gwt.comet.server.GwtAtmosphereResource;
import com.greencat.gwt.comet.server.GwtAtmosphereServlet;
import com.greencat.gwt.comet.server.GwtResponseWriter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.atmosphere.cpr.AtmosphereEventLifecycle;
import org.atmosphere.cpr.AtmosphereResource;
import org.atmosphere.cpr.AtmosphereResourceEvent;
import org.atmosphere.cpr.AtmosphereResourceEventListener;
import org.atmosphere.cpr.Broadcaster;
import org.atmosphere.cpr.Serializer;

/**
 *
 * @author p.havelaar
 */
public class GwtAtmosphereResourceImpl implements GwtAtmosphereResource {

    public GwtAtmosphereResourceImpl(AtmosphereResource<HttpServletRequest, HttpServletResponse> resource,
            GwtAtmosphereServlet servlet, int heartBeatInterval) throws IOException {
        this.servlet = servlet;
        this.atmResource = resource;
        this.heartBeatInterval = heartBeatInterval;
        this.writer = createResponseWriter();
    }

    @Override
    public Broadcaster getBroadcaster() {
        return atmResource.getBroadcaster();
    }

    @Override
    public HttpSession getSession() {
        return getSession();
    }

    @Override
    public HttpSession getSession(boolean create) {
        return atmResource.getRequest().getSession(create);
    }

    @Override
    public GwtResponseWriter getResponseWriter() {
        return writer;
    }

    @Override
    public int getHeartBeatInterval() {
        return heartBeatInterval;
    }

    @Override
    public void broadcast(Serializable message) throws IOException {
        getBroadcaster().broadcast(message, atmResource);
    }

    @Override
    public void broadcast(List<Serializable> messages) throws IOException {
        getBroadcaster().broadcast(messages, atmResource);
    }

    @Override
    public AtmosphereResource<HttpServletRequest, HttpServletResponse> getAtmosphereResource() {
        return atmResource;
    }

    /**
     * Check to see if this atmosphere resource is still in use by the system.
     * It will query the associated broadcaster if it is still referenced.
     *
     * @param resource
     * @return
     */
    @Override
    public boolean isAlive() {
        if (atmResource.getBroadcaster() == null) {
            return false;
        }
        Iterator<AtmosphereResource> it = atmResource.getBroadcaster().getAtmosphereResources();
        while (it.hasNext()) {
            if (it.next() == atmResource) {
                return true;
            }
        }
        return false;
    }

    @Override
    public long getConnectionID() {
        return writer.connectionID;
    }

    public void suspend() throws IOException {
        if (!suspended) {
            atmResource.setSerializer(serializer);
            if (atmResource instanceof AtmosphereEventLifecycle) {
                AtmosphereEventLifecycle ael = (AtmosphereEventLifecycle)atmResource;
                ael.addEventListener(eventListener);
            }
            writer.suspend();
            atmResource.suspend();
            suspended = true;
        }
    }

    @Override
    public <T> void setAttribute(String name, T value) {
        atmResource.getRequest().setAttribute(name, value);
    }

    @Override
    public <T> T getAttribute(String name) {
        return (T) atmResource.getRequest().getAttribute(name);
    }

    public GwtResponseWriterImpl getWriterImpl() {
        return writer;
    }
    
	ScheduledFuture<?> scheduleHeartbeat() {
        return getBroadcaster().getBroadcasterConfig().getScheduledExecutorService()
                .schedule(heartBeatTask, heartBeatInterval, TimeUnit.MILLISECONDS);
	}

    void terminate(boolean serverInitiated) {
        GwtAtmosphereServlet s = servlet;
        if (s != null) {
            servlet = null;
            if (suspended) {
                atmResource.resume();
            }
            s.cometTerminated(this, serverInitiated);
        }
    }

	private GwtResponseWriterImpl createResponseWriter() throws IOException {

		ClientOracle clientOracle = GwtUtil.getClientOracle(atmResource.getRequest(), servlet.getServletContext());
		SerializationPolicy serializationPolicy = clientOracle == null ? GwtUtil.createSerializationPolicy() : null;

		String accept = atmResource.getRequest().getHeader("Accept");
		String userAgent = atmResource.getRequest().getHeader("User-Agent");
		if ("text/plain".equals(accept)) {
			return new HTTPRequestResponseWriter(this, serializationPolicy, clientOracle);
		}
		else if (userAgent != null && userAgent.contains("Opera")) {
			return new OperaEventSourceResponseWriter(this, serializationPolicy, clientOracle);
		}
		else {
			return new IEHTMLFileResponseWriter(this, serializationPolicy, clientOracle);
		}
	}

    private final GwtResponseWriterImpl writer;
    private AtmosphereResource<HttpServletRequest, HttpServletResponse> atmResource;
    private final int heartBeatInterval;
    private GwtAtmosphereServlet servlet;
    private boolean suspended = false;
    private Log logger = LogFactory.getLog(getClass());

    private Runnable heartBeatTask = new Runnable() {
        @Override
        public void run() {
            if (isAlive()) {
                try {
                    writer.heartbeat();
                } catch (IOException e) {
                    if (writer.isTerminated()) {
                        logger.debug("Failed heartbeat, connection terminated");
                    } else {
                        logger.warn("Failed heartbeat", e);
                    }
                }
            }
        }
    };

    private final Serializer serializer = new Serializer() {
        @Override
        public void write(OutputStream out, Object o) throws IOException {
            if (o instanceof Serializable) {
                try {
                    writer.write((Serializable)o);
                } catch (IOException e) {
                    if (writer.isTerminated()) {
                        logger.debug("broadcast failed, connection terminated");
                    } else {
                        throw e;
                    }
                }
            } else  if (o instanceof List) {
                List<?> list = (List)o;
                if (list.size() > 0) {
                    if (!(list.get(0) instanceof Serializable)) {
                        throw new IOException("Failed to write a list of objects that are not serializable");
                    }
                    writer.write((List<Serializable>)o);
                }
            } else {
                throw new IOException("Failed to write an object that is not serializable");
            }
        }
    };

    private final AtmosphereResourceEventListener eventListener = new AtmosphereResourceEventListener() {

        @Override
        public void onResume(AtmosphereResourceEvent<HttpServletRequest, HttpServletResponse> event) {
            suspended = false;
            writer.setTerminated(false);
        }

        @Override
        public void onDisconnect(AtmosphereResourceEvent<HttpServletRequest, HttpServletResponse> event) {
            suspended = false;
            writer.setTerminated(false);
        }

        @Override
        public void onBroadcast(AtmosphereResourceEvent<HttpServletRequest, HttpServletResponse> event) {
        }
    };
}
